package com.intct.func;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import com.intct.hbase.bean.AdsConfigBean;
import com.intct.hbase.bean.LoucsBean;
import com.intct.util.JDBCUtil;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;

import java.sql.Connection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author gufg
 * @since 2025-08-21 11:52
 */
public class AdsLocusFunc extends RichMapFunction<LoucsBean, JSONObject> {
    private Map<String, AdsConfigBean> configMap = new HashMap<>();

    @Override
    public void open(Configuration parameters) throws Exception {
        Connection connection = JDBCUtil.getJDBCConnect();
        List<AdsConfigBean> dimConfigBeanList =
                JDBCUtil.getQuery(connection, "select * from travel_config.ads_table_config", AdsConfigBean.class);
        for (AdsConfigBean adsConfigBean : dimConfigBeanList) {
            configMap.put(adsConfigBean.getSourceTable(), adsConfigBean);
        }
        if (connection != null) {
            connection.close();
        }
    }

    @Override
    public JSONObject map(LoucsBean loucsBean) throws Exception {
        // 声明json object
        JSONObject jsonObj = new JSONObject();

        // 配置将JavaBean字段名，从小驼峰转为蛇形命名
        SerializeConfig serializeConfig = new SerializeConfig();
        serializeConfig.setPropertyNamingStrategy(PropertyNamingStrategy.SnakeCase);

        // 数据追加到json object
        jsonObj.put("after", JSON.toJSON(loucsBean, serializeConfig));

        AdsConfigBean adsConfigBean = configMap.get("locus");
        jsonObj.put("sinkTable", adsConfigBean.getSinkTable());
        jsonObj.put("sinkFamily", adsConfigBean.getSinkFamily());
        jsonObj.put("sinkRowKey", adsConfigBean.getSinkRowKey());
        jsonObj.put("op", "r");

        return jsonObj;
    }
}
